package cn.ac.iie.di.rocketmq;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.RESessionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSessionBuilder;
import cn.ac.iie.di.global.GlobalParas;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.log4j.Logger;

/**
 *
 * @author YG
 */
public class SendToAbandon implements Runnable {

        public static Logger log = Logger.getLogger(SendToAbandon.class.getName());
        RESendSession session = null;
        RESendSessionBuilder builder = null;
        String msgType = null;
        String mqName = GlobalParas.sendMq[1].split("[:]")[0];
        String group = null;

        private final long WAITING_TIME = GlobalParas.waittingTime;

        private static AtomicInteger count = new AtomicInteger(0);

//    public RocketMqProducer(String mq2group) {
//        mqName = mq2group.split("[:]")[0];
//        group = mq2group.split("[:]")[1];
//    }
        @Override
        public void run() {
                int addcount = 0;
                long startTime = System.currentTimeMillis();
                while (true) {
                        try {
                                HashMap map = GlobalParas.sendMsgPool2.poll(20, TimeUnit.MILLISECONDS);
                                if (map == null) {
                                        if (System.currentTimeMillis() - startTime >= WAITING_TIME && addcount > 0) {
                                                if (session != null) {
                                                        try {
                                                                session.flush();
                                                                log.info("send to " + mqName + " success " + addcount);
                                                                addcount = 0;
                                                                continue;
                                                        } catch (Exception ex) {
                                                                log.error(ex, ex);
                                                        }
                                                }
                                        }
                                        if (GlobalParas.killsendMsg && addcount > 0) {
                                                if (session != null) {
                                                        try {
                                                                session.flush();
                                                                log.info("send to " + mqName + " success " + addcount);
                                                                addcount = 0;
                                                                continue;
                                                        } catch (Exception ex) {
                                                                log.error(ex, ex);
                                                        }
                                                }
                                                break;
                                        }
                                        if (GlobalParas.killsendMsg) {
                                                if (addcount > 0) {
                                                        session.flush();
                                                        log.info("send to " + mqName + " success " + addcount);
                                                }
                                                break;
                                        }
                                } else {
                                        if (builder == null) {
                                                try {
                                                        String sendMqString = mqName;
                                                        builder = (RESendSessionBuilder) GlobalParas.conn.getSendSessionBuilder(sendMqString);
                                                } catch (REConnectionException ex) {
                                                }
                                        }
                                        if (session == null) {
                                                Iterator it = map.entrySet().iterator();
                                                while (it.hasNext()) {
                                                        Map.Entry entry = (Map.Entry) it.next();
                                                        String column = (String) entry.getKey();
                                                        LinkedList list = (LinkedList) (List) map.get(column);
                                                        builder.addColumn(column, (REFieldType) list.get(0), true);
                                                }
                                                try {
                                                        session = (RESendSession) builder.build();
                                                        session.start();
                                                } catch (Exception ex) {
                                                        log.error(ex, ex);
                                                }
                                        }

                                        Iterator it = map.entrySet().iterator();

                                        while (it.hasNext()) {
                                                Map.Entry entry = (Map.Entry) it.next();
                                                String column = (String) entry.getKey();
                                                List list = (List) map.get(column);
                                                try {
                                                        switch ((REFieldType) list.get(0)) {
                                                                case String:
                                                                        session.setString(column, String.valueOf(list.get(1)));
                                                                        break;
                                                                case Int:
                                                                        session.setInt(column, Integer.valueOf(list.get(1).toString()));
                                                                        break;
                                                                case Long:
                                                                        session.setLong(column, Long.valueOf(list.get(1).toString()));
                                                                        break;
                                                                case Float:
                                                                        session.setFloat(column, Float.valueOf(list.get(1).toString()));
                                                                        break;
                                                                case Double:
                                                                        session.setDouble(column, Double.valueOf(list.get(1).toString()));
                                                                        break;
                                                                case Binary:
                                                                        session.setBinary(column, list.get(1).toString().getBytes());
                                                                        break;
                                                                case Longs:
                                                                        session.setLongs(column, (long[]) list.get(1));
                                                                        break;
                                                                case Ints:
                                                                        session.setInts(column, (int[]) list.get(1));
                                                                        break;
                                                                case Strings:
                                                                        session.setStrings(column, (String[]) list.get(1));
                                                                        break;
                                                                case Boolean:
                                                                        session.setBoolean(column, (boolean) list.get(1));
                                                                        break;
                                                                case Booleans:
                                                                        session.setBooleans(column, (boolean[]) list.get(1));
                                                                        break;
                                                                case Struct:
                                                                        session.setStruct(column, (Map) list.get(1));
                                                                        break;
                                                                case Structs:
                                                                        session.setStructs(column, (Map[]) list.get(1));
                                                                        break;
                                                                default:
                                                                        log.error("There is not the type of ");
                                                        }
                                                } catch (Exception ex) {
                                                        log.error(ex, ex);
                                                }
                                        }
                                        try {
                                                session.add();
                                                addcount++;
                                        } catch (RESessionException ex) {
                                                log.error(ex, ex);
                                                while (true) {
                                                        try {
                                                                //如果添加失败
                                                                //则将之前添加的部分先flush，将计数值置0
                                                                if (addcount > 0) {
                                                                        session.flush();
                                                                        log.info("send to " + mqName + " success " + addcount);
                                                                        addcount = 0;
                                                                        break;
                                                                }
                                                        } catch (Exception e) {
                                                        }
                                                }
                                                //然后再添加一次
                                                session.add();
                                                addcount++;
                                        }

                                        Long time = System.currentTimeMillis() - startTime;
                                        if (time < WAITING_TIME) {
                                                if (addcount >= GlobalParas.bagSize) {
                                                        session.flush();
                                                        log.info("send to " + mqName + " success " + addcount);
                                                        addcount = 0;
                                                        startTime = System.currentTimeMillis();
                                                }
                                        } else if (addcount > 0) {
                                                session.flush();
                                                log.info("send to " + mqName + " success " + addcount);
                                                addcount = 0;
                                                startTime = System.currentTimeMillis();
                                        }
                                        if (GlobalParas.killsendMsg) {
                                                if (addcount > 0) {
                                                        session.flush();
                                                        log.info("send to " + mqName + " success " + addcount);
                                                }
                                                break;
                                        }
                                }
                        } catch (Exception ex) {
                                log.error(ex, ex);
                        }
                }
                if (!GlobalParas.MqSendQueue.isEmpty()) {
                        GlobalParas.MqSendQueue.poll();
                        log.info("GlobalParas.MqProducerQueue size is : " + GlobalParas.MqSendQueue.size());
                }
                log.info("RocketMq producer exit ");
        }
}
